
import com.huaban.analysis.jieba.JiebaSegmenter;
import org.apache.spark.sql.*;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.api.java.UDF2;
import org.apache.spark.sql.types.DataTypes;
import scala.Int;


import java.io.File;
import java.sql.Array;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class Cleaning {

    static SparkSession session = SparkSession.builder().master("local").appName("cleaning").getOrCreate();
    static String url = "jdbc:mysql://81.70.38.202:3306/PositionInfo?autoReconnect=true&useUnicode=true&characterEncoding=utf8";
    static Properties properties = new Properties();

    public static void main(String[] args) throws AnalysisException {
//        定义插入列udf函数
        session.udf().register("addCol", new UDF2<String, String, String>() {
            @Override
            public String call(String olds, String news) throws Exception {
                return news;
            }
        }, DataTypes.StringType);

        properties.setProperty("user", "root");
        properties.setProperty("password", "795597");

//        获取路径下的文件名（不递归）
        String path = "../data";
        File file = new File(path);
        String[] list = file.list();

//        创建一个map集合存放读取的数据，键为城市名称
        HashMap<String, Dataset<Row>> fileMap = new HashMap<>();
        assert list != null;
        for (String name : list) {
            fileMap.put(
                    name.split("\\.")[0],
                    session.read().option("header", "true").option("multiLine", true).csv("../data/" + name)
            );
        }

        fileMap.forEach((s, dataset) -> {
            try {

                eduExpSal(dataset, s);
//                DistrictCountToMysql(dataset, s);
//                wordCount(dataset, s);
//                companyTypeCount(dataset, s);
//                companySizeDataProcessing(dataset, s);
//                situationCount(dataset, s);

            } catch (Exception e) {
                e.printStackTrace();
            }
        });


//        测试
        Dataset<Row> liouZhou = fileMap.get("广州");

        Dataset<Row> drop = liouZhou.drop("标题链接", "图片", "jobcardleft_链接", "信息");

        drop.select("关键词1").show();
        eduExpSal(liouZhou,"广州");


    }

    static void companyTypeCount(Dataset<Row> dataset, String name) {
        Dataset<Row> select = dataset.select("关键词1");
        Dataset<String> map = select.map(row -> {
            String regEx = "(?=[\u4e00-\u9fa5]).+?(?=[未已不天A-Z0-9])";
            Pattern p = Pattern.compile(regEx);
            Matcher m = p.matcher(row.toString());
            if (m.find()) {
                return m.group().matches("([未已不天A-Z]).*")?null:m.group();
            }
            return null;
        }, Encoders.STRING());
        Dataset<Row> result = map.groupBy("value").count().where("value != 'null'")
                .selectExpr("addCol(value,'" + name + "') as city", "value as type", "count");
//        result.write().mode(SaveMode.Append).jdbc(url, "company_type", properties);
        result.show();
    }

    static void wordCount(Dataset<Row> dataset, String name) {
        Dataset<String> segmentationWord = dataset.select("关键词2").flatMap(row -> {
            String regEx = "HTML|html|css|CSS|([Mm]|[Nn])[YyOo](DB|SQL)|SQL|[A-Z]([a-z]+)(DB)|git|GIT|IDEA|JS|PHP|SpringMVC|[sS]pring[BbCc](oot|loud)|([Jj])ava([Ss])cript|[A-Z]{2,10}|(?=[A-Z]).+?(?=[^a-z])|[\u4e00-\u9fa5]+|[a-z]+";
            Pattern p = Pattern.compile(regEx);
            Matcher m = p.matcher(row.toString());
            ArrayList<String> arrayList = new ArrayList<>();
            while (m.find()) {
                if (m.group().length() > 1 || "c".equals(m.group()) || "C".equals(m.group())) {
                    arrayList.add(m.group());
                }
            }
            return arrayList.iterator();
        }, Encoders.STRING());
        Dataset<Row> result = segmentationWord.groupBy("value").count()
                .selectExpr("addCol(value,'" + name + "') as city", "value as key_word", "count");
        result.show();

        result.write().mode(SaveMode.Append).jdbc(url,"word_count",properties);
    }

    static void situationCount(Dataset<Row> dataset, String name) {
        Dataset<Row> select = dataset.select("关键词1");
        Dataset<Row> result = select.map(row -> {
                    String regEx = "(?=[未已不天A-Z]).+?(?=[0-9])";
                    Pattern p = Pattern.compile(regEx);
                    Matcher m = p.matcher(row.toString());
                    if (m.find()) {
                        return m.group().length() > 1 ? m.group() : null;
                    }
                    return null;
                }, Encoders.STRING())
                .groupBy("value").count().where("value != 'null'")
                .selectExpr("addCol(value,'" + name + "') as city", "value as situation", "count");
//        result.write().mode(SaveMode.Append).jdbc(url, "company_situation", properties);
        result.show();
    }

    static void companySizeDataProcessing(Dataset<Row> dataset, String name) {
        Dataset<Row> select = dataset.select("关键词1");

        Dataset<Row> result = select.map(row -> {
                    String regEx = "[0-9]+";
                    Pattern p = Pattern.compile(regEx);
                    Matcher m = p.matcher(row.toString());
                    ArrayList<String> list = new ArrayList<>();
                    StringBuilder s = new StringBuilder();
                    while (m.find()) {
                        list.add(m.group());
                    }
                    switch (list.size()) {
                        case 1:
                            s.append(list.get(0)).append("以上");
                            break;
                        case 2:
                            s.append(list.get(0)).append("-").append(list.get(1));
                            break;
                        default:
                            s.append(list.get(1)).append("-").append(list.get(2));
                            break;
                    }
                    return s.toString();
                }, Encoders.STRING())
                .groupBy("value").count()
                .selectExpr("addCol(value,'" + name + "') as city", "value as people", "count");
//        result.write().mode(SaveMode.Append).jdbc(url, "company_people", properties);
        result.show();
    }

    static void eduExpSal(Dataset<Row> dataset, String name) {
        session.udf().register("extractNum", new UDF1<String, Float>() {
            @Override
            public Float call(String s) throws Exception {
                String regEx = "[^0-9]";
                Pattern p = Pattern.compile(regEx);
                Matcher m = p.matcher(s);
                String s1 = m.replaceAll("-").split("-")[0];
                if (s.contains("K") || s.contains("k")) {
                    return Float.parseFloat(s1) * 1000;
                }
                return Float.parseFloat(s1) * 30;
            }
        }, DataTypes.FloatType);
        String[] qualifications = {"不限","初中","中专","高中","大专", "本科", "硕士","博士"};
        String[] experiences = {"不限","应届","1年以内","1-3年","3-5年","5-10年","10年以上"};
        session.udf().register("myFilterQ", new UDF1<String, String>() {
            @Override
            public String call(String s) throws Exception {
                for (String qualification : qualifications) {
                    if (s.contains(qualification)) {
                        return qualification;
                    }
                }
                return null;
            }
        }, DataTypes.StringType);
        session.udf().register("myFilterE", new UDF1<String, String>() {
            @Override
            public String call(String s) throws Exception {
                for (String experience : experiences) {
                    if (s.contains(experience)) {
                        return experience;
                    }
                }
                return null;
            }
        }, DataTypes.StringType);
        Dataset<Row> selectExpr = dataset.selectExpr(
                "extractNum(`薪水`) as salary", "split(`关键词`,'\n')[0] as experience", "split(`关键词`,'\n')[1] as qualifications"
        );
        Dataset<Row> groupSalaryMean = selectExpr.groupBy("qualifications", "experience").mean("salary");

        Dataset<Row> result = groupSalaryMean.selectExpr(
                "addCol(experience,'" + name + "') as city",
                "myFilterQ(qualifications) as qualifications",
                "myFilterE(experience) as experience",
                "int(`avg(salary)`) as salary"
        ).where("qualifications != 'null' or experience != 'null'");
//        result.show();
        result.write().mode(SaveMode.Append).jdbc(url,"edu_exp_salary",properties);
    }

    static void DistrictCountToMysql(Dataset<Row> dataset, String name) throws AnalysisException {
        session.udf().register("extractDistrict", new UDF1<String, String>() {
            @Override
            public String call(String s) throws Exception {
                String[] split = s.split("·");
                return split.length > 1 ? split[1] : split[0];
            }
        }, DataTypes.StringType);

        Dataset<Row> selectExpr = dataset.selectExpr("extractDistrict(jobarea) as district");
        Dataset<Row> count = selectExpr.where("district != '" + name + "'").groupBy("district").count();
        Dataset<Row> result = count.selectExpr("addCol(district,'" + name + "') as city", "district", "count");
        result.show();

//        result.write().mode(SaveMode.Append).jdbc(url,"district_count",properties);

    }
}
